package com.ke.searchhouse.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.google.common.primitives.Longs;
import com.ke.searchhouse.base.HouseSort;
import com.ke.searchhouse.base.RentValueBlock;
import com.ke.searchhouse.dto.HouseBucketDTO;
import com.ke.searchhouse.dto.HouseDTO;
import com.ke.searchhouse.entity.*;
import com.ke.searchhouse.form.MapSearch;
import com.ke.searchhouse.form.RentSearch;
import com.ke.searchhouse.repository.HouseDetailRepository;
import com.ke.searchhouse.repository.HouseRepository;
import com.ke.searchhouse.repository.HouseTagRepository;
import com.ke.searchhouse.repository.SupportAddressRepository;
import com.ke.searchhouse.service.IAddressService;
import com.ke.searchhouse.service.ISearchService;
import com.ke.searchhouse.template.BaiduMapLocation;
import com.ke.searchhouse.template.HouseIndexKey;
import com.ke.searchhouse.template.HouseIndexTemplate;
import com.ke.searchhouse.util.JsonMapperUtil;
import com.ke.searchhouse.util.ServiceMultiResult;
import com.ke.searchhouse.util.ServiceResult;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.lucene.queryparser.xml.builders.BooleanQueryBuilder;
import org.apache.lucene.search.BooleanQuery;
import org.codehaus.jackson.type.TypeReference;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeAction;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeRequest;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeRequestBuilder;
import org.elasticsearch.action.admin.indices.analyze.AnalyzeResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.geo.GeoPoint;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryAction;
import org.elasticsearch.index.reindex.DeleteByQueryRequestBuilder;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
import org.elasticsearch.search.suggest.completion.CompletionSuggestionBuilder;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

@Service
@Slf4j
public class SearchServiceImpl implements ISearchService {
    private static final String INDEX_NAME = "searchhouse";

    private static final String INDEX_TYPE = "house";

    private static final String INDEX_TOPIC = "house_build";

    @Autowired
    private HouseRepository houseRepository;
    @Autowired
    private TransportClient esClient;
    @Autowired
    private HouseDetailRepository houseDetailRepository;
    @Autowired
    private HouseTagRepository tagRepository;
    @Autowired
    private SupportAddressRepository supportAddressRepository;
    @Autowired
    private IAddressService addressService;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @KafkaListener(topics = INDEX_TOPIC)
    public void handleMessage(String content) {
        HouseIndexMessage message = JsonMapperUtil.string2Obj(content, new TypeReference<HouseIndexMessage>() {
        });

        switch (message.getOperation()) {
            case HouseIndexMessage.INDEX:
                createOrUpdateIndex(message);
                break;
            case HouseIndexMessage.REMOVE:
                removeIndex(message);
                break;
            default:
                log.warn("Not support message operation: {}", message);
                break;
        }
    }

    private void createOrUpdateIndex(HouseIndexMessage message) {
        Long houseId = message.getHouseId();

        log.info("createOrUpdateIndex -> houseId: {}", houseId);

        House house = houseRepository.findById(houseId).orElse(null);

        if (house == null) {
            log.warn("house not exist, houseId: {}", houseId);
            index(houseId, message.getRetry() + 1);
            return;
        }

        HouseIndexTemplate houseIndexTemplate = new HouseIndexTemplate();
        BeanUtils.copyProperties(house, houseIndexTemplate);

        HouseDetail houseDetail = houseDetailRepository.findByHouseId(houseId);

        if (houseDetail == null) {
            // TODO: 异常处理
        }

        BeanUtils.copyProperties(houseDetail, houseIndexTemplate);

        SupportAddress city = supportAddressRepository.
                findByEnNameAndLevel(house.getCityEnName(), SupportAddress.Level.CITY.getValue());

        SupportAddress region = supportAddressRepository.
                findByEnNameAndLevel(house.getRegionEnName(), SupportAddress.Level.REGION.getValue());

        String address = city.getCnName() + region.getCnName() + house.getStreet() + house.getDistrict() + houseDetail.getDetailAddress();

        ServiceResult<BaiduMapLocation> result = addressService.getBaiduMapLocation(city.getCnName(), address);

        if (!result.isSuccess()) {
            log.warn("Get BaiduMap Location error:{}, {}", city.getCnName(), address);
            index(houseId, message.getRetry() + 1);
            return;
        }

        houseIndexTemplate.setLocation(result.getResult());

        List<HouseTag> tags = tagRepository.findAllByHouseId(houseId);

        if (tags != null && !tags.isEmpty()) {
            List<String> tagStr = new ArrayList<>();
            tags.forEach(houseTag -> tagStr.add(houseTag.getName()));
            houseIndexTemplate.setTags(tagStr);
        }

        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, houseIndexTemplate.getHouseId()));

        log.info(requestBuilder.toString());
        SearchResponse response = requestBuilder.get();

        long totalHit = response.getHits().getTotalHits();
        boolean res = false;

        if (totalHit == 0) {
            res = create(houseIndexTemplate);
        } else if (totalHit == 1) {
            String esId = response.getHits().getAt(0).getId();
            res = update(esId, houseIndexTemplate);
        } else {
            res = deleteAndCreate(totalHit, houseIndexTemplate);
        }

        ServiceResult serviceResult = addressService.lbsUpload(houseIndexTemplate.getLocation(),
                house.getStreet() + house.getDistrict(),
                city.getCnName() + region.getCnName() + house.getStreet() + house.getDistrict(),
                message.getHouseId(), house.getPrice(), house.getArea());

        if (!res || !serviceResult.isSuccess()) {
            this.index(message.getHouseId(), message.getRetry() + 1);
        }

        log.info("Index success with house: {}", houseId);
    }

    private void removeIndex(HouseIndexMessage message) {
        Long houseId = message.getHouseId();

        DeleteByQueryRequestBuilder queryRequestBuilder = DeleteByQueryAction.INSTANCE
                .newRequestBuilder(esClient)
                .filter(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, houseId))
                .source(INDEX_NAME);

        log.info("Delete by query for house: {}", houseId);

        BulkByScrollResponse response = queryRequestBuilder.get();
        long deleted = response.getDeleted();

        log.info("Total remove: {}", deleted);

        ServiceResult serviceResult = addressService.removeLbs(message.getHouseId());

        if (deleted <= 0 || !serviceResult.isSuccess()) {
            this.remove(houseId, message.getRetry() + 1);
        }
    }

    @Override
    public void index(Long houseId) {
       this.index(houseId, 0);
    }

    @Override
    public void remove(Long houseId) {
        this.remove(houseId, 0);
    }

    @Override
    public ServiceMultiResult<Long> query(RentSearch rentSearch) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();

        boolQuery.filter(
                QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME, rentSearch.getCityEnName())
        );

        if (StringUtils.isNotBlank(rentSearch.getRegionEnName()) &&
                !"*".equals(rentSearch.getRegionEnName())) {
            boolQuery.filter(
                    QueryBuilders.termQuery(HouseIndexKey.REGION_EN_NAME, rentSearch.getRegionEnName())
            );
        }

        // 面积匹配
        RentValueBlock area = RentValueBlock.matchArea(rentSearch.getAreaBlock());
        if (!RentValueBlock.ALL.equals(area)) {
            RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(HouseIndexKey.AREA);
            if (area.getMax() > 0) {
                rangeQueryBuilder.lte(area.getMax());
            }

            if (area.getMin() > 0) {
                rangeQueryBuilder.gte(area.getMin());
            }

            boolQuery.filter(rangeQueryBuilder);
        }

        // 价格匹配
        RentValueBlock price = RentValueBlock.matchPrice(rentSearch.getPriceBlock());
        if (!RentValueBlock.ALL.equals(price)) {
            RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery(HouseIndexKey.PRICE);
            if (price.getMax() > 0) {
                rangeQueryBuilder.lte(price.getMax());
            }

            if (price.getMin() > 0) {
                rangeQueryBuilder.gte(price.getMin());
            }

            boolQuery.filter(rangeQueryBuilder);
        }

        // 朝向匹配
        if (rentSearch.getDirection() > 0) {
            boolQuery.filter(
                    QueryBuilders.termQuery(HouseIndexKey.DIRECTION, rentSearch.getDirection())
            );
        }

        // 租赁方式匹配
        if (rentSearch.getRentWay() > -1) {
            boolQuery.filter(
                    QueryBuilders.termQuery(HouseIndexKey.RENT_WAY, rentSearch.getRentWay())
            );
        }

        boolQuery.should(
                QueryBuilders.matchQuery(HouseIndexKey.TITLE, rentSearch.getKeywords()).boost(2.0f)
        );

        // 关键词匹配
        boolQuery.should(
                QueryBuilders.multiMatchQuery(rentSearch.getKeywords(),
                        HouseIndexKey.TRAFFIC,
                        HouseIndexKey.DISTRICT,
                        HouseIndexKey.ROUND_SERVICE,
                        HouseIndexKey.SUBWAY_LINE_NAME,
                        HouseIndexKey.SUBWAY_STATION_NAME)
        );


        SearchRequestBuilder requestBuilder = esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .addSort(
                        HouseSort.getSortKey(rentSearch.getOrderBy()),
                        SortOrder.fromString(rentSearch.getOrderDirection())
                )
                .setQuery(boolQuery)
                .setFrom(rentSearch.getStart())
                .setSize(rentSearch.getSize())
                .setFetchSource(HouseIndexKey.HOUSE_ID, null);

        log.info(requestBuilder.toString());

        List<Long> houseIds = new ArrayList<>();
        SearchResponse response = requestBuilder.get();

        log.info("response:{}", response.toString());

        if (response.status() != RestStatus.OK) {
            log.error("Search status is not ok for:{} ", requestBuilder);
            return new ServiceMultiResult<>(0, houseIds);
        }

        for (SearchHit hit : response.getHits()) {
            houseIds.add(Long.valueOf(hit.getSource().get(HouseIndexKey.HOUSE_ID).toString()));
        }

        return new ServiceMultiResult<>(response.getHits().getTotalHits(), houseIds);
    }

    @Override
    public ServiceResult<List<String>> suggest(String prefix) {
        CompletionSuggestionBuilder suggestion = SuggestBuilders.completionSuggestion("suggest")
                .prefix(prefix).size(5);

        SuggestBuilder suggestBuilder = new SuggestBuilder();
        suggestBuilder.addSuggestion("autocomplete", suggestion);

        SearchRequestBuilder requestBuilder = this.esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .suggest(suggestBuilder);

        log.info(requestBuilder.toString());

        SearchResponse response = requestBuilder.get();
        Suggest suggest = response.getSuggest();
        Suggest.Suggestion result = suggest.getSuggestion("autocomplete");

        int maxSuggest = 0;
        Set<String> suggestSet = new HashSet<>();

        u:
            for (Object term : result.getEntries()) {
                if (term instanceof CompletionSuggestion.Entry) {
                    CompletionSuggestion.Entry item = (CompletionSuggestion.Entry) term;

                    if (item.getOptions().isEmpty()) {
                        continue;
                    }

                    for (CompletionSuggestion.Entry.Option option : item.getOptions()) {
                        String tip = option.getText().toString();
                        log.info("tip->" + tip);
                        if (suggestSet.contains(tip)) {
                            continue;
                        }

                        suggestSet.add(tip);
                        maxSuggest ++;

                        if (maxSuggest > 5) {
                            break u;
                        }
                    }
                }
            }

            List<String> suggesions = Lists.newArrayList(suggestSet.toArray(new String[]{}));

            return ServiceResult.of(suggesions);
    }

    @Override
    public ServiceResult<Long> aggregateDistrictHouse(String cityEnName, String regionEnName, String district) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery()
                .filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME, cityEnName))
                .filter(QueryBuilders.termQuery(HouseIndexKey.REGION_EN_NAME, regionEnName))
                .filter(QueryBuilders.termQuery(HouseIndexKey.DISTRICT, district));

        SearchRequestBuilder requestBuilder = esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQuery)
                .addAggregation(
                        AggregationBuilders.terms(HouseIndexKey.AGG_DISTRICT)
                        .field(HouseIndexKey.DISTRICT)
                ).setSize(0);

        log.info(requestBuilder.toString());

        SearchResponse response = requestBuilder.get();
        if (response.status() == RestStatus.OK) {
            Terms terms = response.getAggregations().get(HouseIndexKey.AGG_DISTRICT);

            if (terms != null && !terms.getBuckets().isEmpty()) {
                return ServiceResult.of(terms.getBucketByKey(district).getDocCount());
            }
        } else {
            log.warn("Fail to Aggregate to: {}", HouseIndexKey.AGG_DISTRICT);
        }

        return ServiceResult.of(0L);
    }

    @Override
    public ServiceMultiResult<HouseBucketDTO> mapAggregate(String cityEnName) {
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        queryBuilder.filter(
                QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME, cityEnName)
        );

        AggregationBuilder aggregationBuilder = AggregationBuilders
                .terms(HouseIndexKey.AGG_REGION).field(HouseIndexKey.REGION_EN_NAME);


        SearchRequestBuilder requestBuilder = esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(queryBuilder)
                .addAggregation(aggregationBuilder);

        log.info(requestBuilder.toString());

        SearchResponse response = requestBuilder.get();
        List<HouseBucketDTO> buckets = new ArrayList<>();

        if (response.status() != RestStatus.OK) {
            log.warn("Aggreagate status is not ok for: {}", requestBuilder);
            return new ServiceMultiResult<>(0, buckets);
        }

        Terms terms = response.getAggregations().get(HouseIndexKey.AGG_REGION);

        for (Terms.Bucket bucket : terms.getBuckets()) {
            buckets.add(new HouseBucketDTO(bucket.getKeyAsString(), bucket.getDocCount()));
        }

        return new ServiceMultiResult<>(response.getHits().getTotalHits(), buckets);
    }

    @Override
    public ServiceMultiResult<Long> mapQuery(String cityEnName, String orderBy, String orderDirection, int start, int size) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME, cityEnName));

        SearchRequestBuilder searchRequestBuilder = esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQuery)
                .addSort(HouseSort.getSortKey(orderBy), SortOrder.fromString(orderDirection))
                .setFrom(start)
                .setSize(size);

        List<Long> houseIds = new ArrayList<>();
        SearchResponse response = searchRequestBuilder.get();

        if (response.status() != RestStatus.OK) {
            log.warn("Search status is not ok for: {}", searchRequestBuilder);
            return new ServiceMultiResult<>(0, new ArrayList<>());
        }

        for (SearchHit hit : response.getHits()) {
            houseIds.add(Longs.tryParse(String.valueOf(hit.getSource().get(HouseIndexKey.HOUSE_ID))));
        }

        return new ServiceMultiResult<>(response.getHits().getTotalHits(), houseIds);
    }

    @Override
    public ServiceMultiResult<Long> mapQuery(MapSearch mapSearch) {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        boolQuery.filter(QueryBuilders.termQuery(HouseIndexKey.CITY_EN_NAME, mapSearch.getCityEnName()));

        boolQuery.filter(
                QueryBuilders.geoBoundingBoxQuery(HouseIndexKey.LOCATION)
                .setCorners(
                        new GeoPoint(mapSearch.getLeftLatitude(), mapSearch.getLeftLongitude()),
                        new GeoPoint(mapSearch.getRightLatitude(), mapSearch.getRightLongitude())
                )
        );

        SearchRequestBuilder searchRequestBuilder = esClient.prepareSearch(INDEX_NAME)
                .setTypes(INDEX_TYPE)
                .setQuery(boolQuery)
                .addSort(HouseSort.getSortKey(mapSearch.getOrderBy()), SortOrder.fromString(mapSearch.getOrderDirection()))
                .setFrom(mapSearch.getStart())
                .setSize(mapSearch.getSize());

        List<Long> houseIds = new ArrayList<>();
        SearchResponse response = searchRequestBuilder.get();

        if (response.status() != RestStatus.OK) {
            log.warn("Search status is not ok for: {}", searchRequestBuilder);
            return new ServiceMultiResult<>(0, new ArrayList<>());
        }

        for (SearchHit hit : response.getHits()) {
            houseIds.add(Longs.tryParse(String.valueOf(hit.getSource().get(HouseIndexKey.HOUSE_ID))));
        }

        return new ServiceMultiResult<>(response.getHits().getTotalHits(), houseIds);
    }

    private void index(Long houseId, Integer retryCount) {
        if (retryCount > HouseIndexMessage.MAX_RETRY) {
            log.error("Retry index time over {} for house: {}, please check it!", HouseIndexMessage.MAX_RETRY, houseId);
            return;
        }

        HouseIndexMessage message = new HouseIndexMessage(houseId, HouseIndexMessage.INDEX, retryCount);

        kafkaTemplate.send(INDEX_TOPIC, JsonMapperUtil.obj2String(message));
    }


    private void remove(Long houseId, Integer retryCount) {
        if (retryCount > HouseIndexMessage.MAX_RETRY) {
            log.error("Retry remove time over {} for house: {}, please check it!", HouseIndexMessage.MAX_RETRY, houseId);
            return;
        }

        HouseIndexMessage message = new HouseIndexMessage(houseId, HouseIndexMessage.REMOVE, retryCount);

        kafkaTemplate.send(INDEX_TOPIC, JsonMapperUtil.obj2String(message));
    }



    private Boolean create(HouseIndexTemplate houseIndexTemplate) {
        if (!updateSuggest(houseIndexTemplate)) {
            return false;
        }

        log.info("Create index with house: {}", houseIndexTemplate.getHouseId());

        log.info("obj2String->{}", JsonMapperUtil.obj2String(houseIndexTemplate));

         IndexResponse response = esClient.prepareIndex(INDEX_NAME, INDEX_TYPE)
                 .setSource(JsonMapperUtil.obj2String(houseIndexTemplate), XContentType.JSON).get();


         if (response.status() == RestStatus.CREATED) {
             return true;
         }

         return false;
    }


    private Boolean update(String esId, HouseIndexTemplate houseIndexTemplate) {
        if (!updateSuggest(houseIndexTemplate)) {
            return false;
        }

        log.info("Update index with house: {}, esId: {}", houseIndexTemplate.getHouseId(), esId);

        UpdateResponse response = esClient.prepareUpdate(INDEX_NAME, INDEX_TYPE, esId)
                .setDoc(JsonMapperUtil.obj2String(houseIndexTemplate), XContentType.JSON).get();

        if (response.status() == RestStatus.OK) {
            return true;
        }

        return false;
    }

    private Boolean deleteAndCreate(Long totalHit, HouseIndexTemplate houseIndexTemplate) {
        DeleteByQueryRequestBuilder queryRequestBuilder = DeleteByQueryAction.INSTANCE
                .newRequestBuilder(esClient)
                .filter(QueryBuilders.termQuery(HouseIndexKey.HOUSE_ID, houseIndexTemplate.getHouseId()))
                .source(INDEX_NAME);

        log.info("Delete by query for house: {}", houseIndexTemplate.getHouseId());

        BulkByScrollResponse response = queryRequestBuilder.get();
        long deleted = response.getDeleted();

        if (totalHit != deleted) {
            log.warn("Need deleted: {}, but {} was deleted", totalHit, deleted);
            return false;
        }

        return create(houseIndexTemplate);
    }

    private boolean updateSuggest(HouseIndexTemplate indexTemplate) {
        AnalyzeRequestBuilder requestBuilder = new AnalyzeRequestBuilder(esClient, AnalyzeAction.INSTANCE, INDEX_NAME,
                indexTemplate.getTitle(), indexTemplate.getLayoutDesc(), indexTemplate.getRoundService(),
                indexTemplate.getDescription(), indexTemplate.getSubwayLineName(), indexTemplate.getSubwayStationName());

        requestBuilder.setAnalyzer("ik_smart");

        AnalyzeResponse response  = requestBuilder.get();
        List<AnalyzeResponse.AnalyzeToken> tokens = response.getTokens();

        if (tokens == null || tokens.size() == 0) {
            log.warn("cannot analyze token for house: {}", indexTemplate.getHouseId());
            return false;
        }

        List<HouseSuggest> houseSuggests = new ArrayList<>();
        for (AnalyzeResponse.AnalyzeToken token : tokens) {
            // 去除数字类型和小于两个字符的结果
            if ("<NUM>".equals(token) || token.getTerm().length() < 2) {
                continue;
            }

            HouseSuggest suggest = new HouseSuggest();
            suggest.setInput(token.getTerm());

            houseSuggests.add(suggest);
        }

        // 定制化小区自动补全
        HouseSuggest suggest = new HouseSuggest();
        suggest.setInput(indexTemplate.getDistrict());
        houseSuggests.add(suggest);

        indexTemplate.setSuggest(houseSuggests);
        return true;
 }


}
